package com.flc.gateway.dataqueue;

import com.flc.gateway.mq.MqProducer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * @author haizi
 * @date 2018/1/2
 */
public class ReportQueue extends Thread {

    private static final Logger logger = LoggerFactory.getLogger(ReportQueue.class);

    public ReportQueue(int capacity, String topic) {
        super.setName("ReportQueue");
        this.dataQueue = new ArrayBlockingQueue<>(capacity);
        this.producer = MqProducer.getProducer();
        this.topic = topic;
    }

    @Override
    public void run() {
        logger.info("开启报告数据队列...");
        if (MqProducer.getState()) {
            logger.info("开启报告数据队列成功！");
        }
        while (MqProducer.getState()) {
            try {
                byte[] data = this.take();
                // ToDo 把取到的数据发往登陆对应的消息队列
                Message message = new Message(topic, data);
                this.producer.send(message);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void put(byte[] data) throws InterruptedException {
        dataQueue.put(data);
    }

    private byte[] take() throws InterruptedException {
        return dataQueue.take();
    }

    private BlockingQueue<byte[]> dataQueue;
    private DefaultMQProducer producer = null;
    private Boolean status = false;
    private String topic;
}
